/**
 * File: AbstractLoadProcessor.java
 * Created by: mhaimel
 * Created on: 10 Jul 2009
 * CVS:  $Id: AbstractLoadProcessor.java,v 1.3 2009/10/09 15:10:12 mhaimel Exp $
 */
package uk.ac.ebi.curtain.processor.load.impl;

import java.io.IOException;
import java.io.Serializable;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import uk.ac.ebi.curtain.processor.AbstractProcessor;
import uk.ac.ebi.curtain.util.FileIndexWriter;
import uk.ac.ebi.curtain.util.file.FastaValidationParser;
import uk.ac.ebi.curtain.util.file.FastqValidationParser;
import uk.ac.ebi.curtain.utils.CurtainUncheckedException;
import uk.ac.ebi.curtain.utils.data.FileType;
import uk.ac.ebi.curtain.utils.file.EntryMapper;
import uk.ac.ebi.curtain.utils.file.EntryMapperBean;
import uk.ac.ebi.curtain.utils.file.FileInfo;
import uk.ac.ebi.curtain.utils.file.ParseUtil;
import uk.ac.ebi.curtain.utils.io.impl.BufferedRafReader;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;
import uk.ac.ebi.velvet.model.Strand;
import uk.ac.ebi.velvet.util.io.SequenceUtil;
import uk.ac.ebi.velvet.util.io.WriteException;

/**
 * @author mhaimel
 *
 */
public abstract class AbstractLoadProcessor<I extends Serializable,O extends Serializable> extends AbstractProcessor<I,O> {
	
	private final AtomicReference<FileIndexWriter> writer = new AtomicReference<FileIndexWriter>();
	private final ReadCollHolder buffer = new ReadCollHolder(1000);

	protected FileInfo getMyInput(){
		return (FileInfo) getInput();
	}
	
	@Override
	public void process() {
		load(getMyInput());
	}
	
	private FileIndexWriter getWriter() {
		return writer.get();
	}
	public void setWriter(FileIndexWriter writer) {
		this.writer.set(writer);
	}

	protected void addRead(FileInfo fileInfo, String key, Strand strand, Long position) {
		if(!buffer.hasSpace()){
			flushBuffer();
		}
		buffer.addRead(getCurrentFileId(fileInfo), key, strand, position);
	}
	
	private void flushBuffer() {
		getWriter().addReadBatch(
				buffer.getFileIds(), 
				buffer.getKeys(),
				buffer.getStrands(), 
				buffer.getPositions());
		buffer.reset();
	}

	private static class ReadCollHolder{
		private final Integer[] fileIds;
		private final String[] keys;
		private final Strand[] strands;
		private final Long[] positions;
		private final int buffsize;
		private final AtomicInteger iPos = new AtomicInteger(0);
		
		public ReadCollHolder(int buffSize){
			this.buffsize = buffSize;
			this.fileIds = new Integer[buffSize];
			this.keys = new String[buffSize];
			this.strands = new Strand[buffSize];
			this.positions = new Long[buffSize]; 
		}		
		
		private boolean hasSpace(){
			return iPos.get() < (buffsize); 
		}
		
		private void addRead(Integer fileId, String key, Strand strand, Long position){
			if(!hasSpace()){
				throw new CurtainUncheckedException("Not able to add -> buffer is full!!! " + iPos + ":" + (buffsize-1));
			}
			this.fileIds[iPos.get()] = fileId;
			this.keys[iPos.get()] = key;
			this.strands[iPos.get()] = strand;
			this.positions[iPos.get()] = position;
			iPos.incrementAndGet();
		}	
		
		
		
		public Integer[] getFileIds() {
			return getReturn(fileIds,new Integer[iPos.get()]);
		}

		public String[] getKeys() {
			return getReturn(keys,new String[iPos.get()]);
		}
		
		private <T> T[] getReturn(Object[] src, T[] arr){
//				new Object[iPos];
			System.arraycopy(src, 0, arr, 0, iPos.get());
			return (T[]) arr;
		}

		public Strand[] getStrands() {
			return getReturn(strands, new Strand[iPos.get()]);
		}

		public Long[] getPositions() {
			return getReturn(positions,new Long[iPos.get()]);
		}

		private void reset(){
			iPos.set(0);
		}
		
	}

	protected void load(FileInfo info) throws WriteException{
		try{
			currentFileInfo = info;
			currentStartPos = getContext().getGlobalFileIndex().getStartPosition(info);
			currentFileId = getContext().getGlobalFileIndex().getId(info);
			if(null == currentStartPos){
				throw new CurtainUncheckedException("Contig File not registered for global access: " + info);
			}
			DataMapper mapper = new DataMapper();
			getLog().info("Load file " + info);
			if(info.getFileType().equals(FileType.fastq)){
				getLog().debug("Choose New parser");
				load(new MyFastqParser(),info,mapper);
			} else if(info.getFileType().equals(FileType.fasta)){
				getLog().debug("Choose New parser");
				load(new MyFastaParser(),info,mapper);
			} else {
				load(info.getFileType().getParser(),info,mapper);
			}			
			flushBuffer();
			cleanUpPost();
			getLog().debug("Finished processing file " + info);
		} finally{
			/* closed in controller */
		}
	}
	
	protected void cleanUpPost(){
		/* possible clean up for other classes */
	}

	private void load(ParseUtil parser, FileInfo info, DataMapper mapper) {
		BufferedRafReader in = null;
		try {
			in = new FileIO(info.getFile()).getBufferedRafReader(1024*4); // default: 65536
			parser.parse(in, info, mapper);
		} catch (IOException e) {
			throw new CurtainUncheckedException(e);
		}finally{
			FileIO.closeQuietly(in);
		}
	}
	
	private FileInfo currentFileInfo = null;
	private Long currentStartPos = null;
	private Integer currentFileId = null;
	
	protected Long getCurrentStartPos(FileInfo fi){
		_ensureSameFile(fi);
		return this.currentStartPos;
	}	
	
	protected Integer getCurrentFileId(FileInfo fi){
		_ensureSameFile(fi);
		return this.currentFileId;
	}

	private void _ensureSameFile(FileInfo fi) {
		if(!fi.equals(this.currentFileInfo)){
			throw new CurtainUncheckedException("Entry from unexpected File: " + fi + "; expected: " + this.currentFileInfo);
		}
	}
	
	protected abstract boolean isValidEntry(FileInfo info, String id, long filePos);
	
	private class MyFastqParser extends FastqValidationParser{
		
		@Override
		protected boolean skipId(FileInfo info, String id, long filePos) {
			return !isValidEntry(info,id, filePos);
		}		
		
		@Override
		protected EntryMapperBean buildEntryBean(long pos, FileInfo info,String id, String header, StringBuilder seq, String annot,StringBuilder qual) {
			EntryMapperBean bean = new EntryMapperBean();
			bean.setId(id);
			bean.setFilePosition(pos);
			bean.setInfo(info);
			bean.setSequenceLength(buildSequenceLength(seq));
			return bean;
		}
		
		@Override
		protected boolean skipExtraAnnotation() {
			return true;
		}
	}
	private class MyFastaParser extends FastaValidationParser{
		
		@Override
		protected boolean skipId(FileInfo info, String id, long filePos) {
			return !isValidEntry(info,id, filePos);
		}		
		
		@Override
		protected EntryMapperBean buildEntryBean(FileInfo info, long pos,String id, String header, StringBuilder seq) {
			EntryMapperBean bean = new EntryMapperBean();
			bean.setId(id);
			bean.setFilePosition(pos);
			bean.setInfo(info);
			bean.setSequenceLength(buildSequenceLength(seq));
			return bean;
		}
	}
	

	private int buildSequenceLength(StringBuilder seq) {
		int iCnt = 0;
		char c=' ';
		for(int i = 0; i < seq.length(); ++i){
			c = seq.charAt(i);
			if(SequenceUtil.isDNA(c)){
				++iCnt;
			}
		}
		return iCnt;
	}
	
	protected abstract void addEntry(EntryMapperBean data);
	
	private class DataMapper implements EntryMapper{

		public void mapEntry(EntryMapperBean data) {
			addEntry(data); 				
		}

	}
}
